package com.gitee.feizns.quickstart.mqtt;

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.ArrayList;
import java.util.List;

/**
 * MQTT 连接
 * @author feizns
 * @since 2024/6/21
 */
@Getter
public class MqttConnection implements AutoCloseable {

    /**
     * 客户端 ID
     */
    private final String clientId;

    /**
     * 选项
     */
    private final MqttConnectOptions options;

    /**
     * MQTT 客户端
     */
    private final MqttClient mqttClient;

    /**
     * MQTT 回调
     */
    private final MqttCallbackListAdapter mqttCallbackListAdapter;

    /**
     * MQTT 连接
     * @param host 主机
     * @param username 用户名
     * @param password 密码
     */
    public MqttConnection(String host, String username, String password) throws MqttException {
        this(host, defaultOptions(username, password));
    }

    /**
     * MQTT 连接
     * @param host 主机
     * @param options 选项
     * @throws MqttException MQTT 异常
     */
    public MqttConnection(String host, MqttConnectOptions options) throws MqttException {
        this(defaultMqttClient(host), options);
    }

    /**
     * MQTT 连接
     * @param mqttClient MQTT 客户端
     * @param options 选项
     * @throws MqttException MQTT 异常
     */
    public MqttConnection(MqttClient mqttClient, MqttConnectOptions options) throws MqttException {
        this.options = options;
        this.mqttClient = mqttClient;
        this.mqttCallbackListAdapter = new MqttCallbackListAdapter();
        mqttClient.setCallback(mqttCallbackListAdapter);
        mqttClient.connect(options);
        this.clientId = mqttClient.getClientId();
    }

    @Override
    public void close() throws Exception {
        mqttClient.disconnect();
        mqttClient.close();
    }

    /**
     * 向某个主题发布消息 默认qos：1
     *
     * @param topic:发布的主题
     * @param msg：发布的消息
     */
    public void pub(String topic, String msg) throws MqttException {
        pub(topic, new MqttMessage(msg.getBytes()));
    }

    /**
     * 向某个主题发布消息
     *
     * @param topic: 发布的主题
     * @param msg:   发布的消息
     * @param qos:   消息质量    Qos：0、1、2
     */
    public void pub(String topic, String msg, int qos) throws MqttException {
        pub(topic, msg(msg, qos));
    }


    /**
     * 向某个主题发布消息 默认qos：1
     *
     * @param topic:发布的主题
     * @param msg：发布的消息
     */
    public void pub(String topic, MqttMessage msg) throws MqttException {
        MqttTopic mqttTopic = mqttClient.getTopic(topic);
        MqttDeliveryToken token = mqttTopic.publish(msg);
        token.waitForCompletion();
    }

    /**
     * 订阅某一个主题 ，此方法默认的的Qos等级为：1
     *
     * @param topic 主题
     */
    public void sub(String topic) throws MqttException {
        mqttClient.subscribe(topic);
    }

    /**
     * 订阅某一个主题，可携带Qos
     *
     * @param topic 所要订阅的主题
     * @param qos   消息质量：0、1、2
     */
    public void sub(String topic, int qos) throws MqttException {
        mqttClient.subscribe(topic, qos);
    }

    /**
     * 默认选项
     * @param username 用户名
     * @param password 密码
     * @return {@link MqttConnectOptions }
     */
    public static MqttConnectOptions defaultOptions(String username, String password) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setAutomaticReconnect(true);//默认：false
        options.setCleanSession(false);//默认：true
        //options.setKeepAliveInterval(20);//默认：60
        return options;
    }

    /**
     * 默认的Mqtt客户端
     * @param host 主机
     * @return {@link MqttConnectOptions }
     * @throws MqttException MQTT 异常
     */
    public static MqttClient defaultMqttClient(String host) throws MqttException {
        return new MqttClient(host, randomClientId(), new MemoryPersistence());
    }

    /**
     * 消息
     * @param msg 消息
     * @param qos QoS（服务质量控制）
     * @return {@link MqttConnectOptions }
     */
    public static MqttMessage msg(String msg, int qos) {
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(qos);
        mqttMessage.setPayload(msg.getBytes());
        return mqttMessage;
    }

    /**
     * 随机客户端 ID
     * @return {@link String }
     */
    public static String randomClientId() {
        return "Mqtt-Con-" + (long) (Math.random() * 100000000000000L);
    }

    /**
     * 注册
     * @param mqttCallback MQTT 回调
     */
    public void register(MqttCallback mqttCallback) {
        mqttCallbackListAdapter.register(mqttCallback);
    }

    /**
     * MQTT 回调列表适配器
     * @author feizns
     * @since 2024/06/23
     */
    @RequiredArgsConstructor
    public static class MqttCallbackListAdapter implements MqttCallback {

        /**
         * 回调
         */
        private final List<MqttCallback> callbacks = new ArrayList<>();

        @Override
        public void connectionLost(Throwable cause) {
            for (MqttCallback mqttCallback : callbacks)
                mqttCallback.connectionLost(cause);
        }

        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            for (MqttCallback mqttCallback : callbacks)
                mqttCallback.messageArrived(topic, message);
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            for (MqttCallback mqttCallback : callbacks)
                mqttCallback.deliveryComplete(token);
        }

        /**
         * 注册
         * @param callback 回调
         */
        public void register(MqttCallback callback) {
            callbacks.add(callback);
        }

    }

}
